package com.google.code.p.restrpc.core.async.impl;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import com.google.code.p.restrpc.AsyncRpcResult;
import com.google.code.p.restrpc.core.async.AsynServiceRequest;
import com.google.code.p.restrpc.core.async.AsyncRequestRejectedException;
import com.google.code.p.restrpc.core.async.AsyncRpcServiceProxy;
import com.google.code.p.restrpc.core.async.AsyncServiceQueue;
import com.google.code.p.restrpc.core.async.AsyncServiceQueueNotStartedException;
import com.google.code.p.restrpc.core.async.AsyncServiceStatus;
import com.google.code.p.restrpc.core.thread.NamedThreadFactory;

/**
 * @author dz
 */
public class DefaultAsyncServiceQueue implements AsyncServiceQueue {
    
    private AtomicBoolean started = new AtomicBoolean(false);
    
    private ExecutorService executorService;
    
    private ConcurrentHashMap<String, AsynServiceRequest> requests = new ConcurrentHashMap<String, AsynServiceRequest>();
    
    /**
     * @param options
     */
    public DefaultAsyncServiceQueue(AsyncServiceQueueOptions options) {
        this.executorService = new ThreadPoolExecutor(options.getCoreWorker(),
                                                      options.getMaxWorker(),
                                                      0L,
                                                      TimeUnit.MILLISECONDS,
                                                      new LinkedBlockingQueue<Runnable>(options.getMaxSize()),
                                                      new NamedThreadFactory("restrpc-async-worker",
                                                                             true));
        
        this.started.set(true);
    }
    
    /**
     * 
     */
    public int size() {
        return requests.size();
    }
    
    /**
     * 
     */
    public void addAsynServiceRequest(final AsynServiceRequest request) {
        if (!started.get()) {
            throw new AsyncServiceQueueNotStartedException();
        }
        
        try {
            executorService.execute(new AsyncRpcServiceProxy(request));
            requests.putIfAbsent(request.getRequestId(), request);
            request.setStatus(AsyncServiceStatus.QUEUED);
        }
        catch (RejectedExecutionException e) {
            throw new AsyncRequestRejectedException(e);
        }
    }
    
    public AsyncRpcResult findById(String id) {
        if (!started.get()) {
            throw new AsyncServiceQueueNotStartedException();
        }
        
        AsynServiceRequest request = requests.get(id);
        if (request == null) {
            return null;
        }
        
        DefaultAsyncRpcResult asyncRpcResult = new DefaultAsyncRpcResult();
        asyncRpcResult.setFinished(false);
        asyncRpcResult.setId(id);
        
        return asyncRpcResult;
    }
    
    void removeAsynServiceRequest(String requestId) {
        requests.remove(requestId);
    }
    
    public void shutdown() {
        if (started.get()) {
            this.executorService.shutdown();
            this.requests.clear();
            this.started.set(false);
        }
    }
    
}
